package com.itcam.elasticsearch.once;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Refresh;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

import java.sql.*;
import java.util.HashMap;
import java.util.Map;

/**
 * @author : Cammy.Wu
 * Description : 
 */

public class MySQLToElasticsearch {
    /**
     * 程序的主入口方法，用于将 MySQL 数据库中的数据导入到 Elasticsearch 中。
     *
     * @param args 命令行参数，在本程序中未使用。
     * @throws Exception 当在连接数据库、执行查询或与 Elasticsearch 交互过程中发生异常时抛出。
     */
    public static void main(String[] args) throws Exception {
        // 连接 Elasticsearch
        // 创建一个低级别 REST 客户端，连接到本地运行在 9200 端口的 Elasticsearch 实例
        RestClient restClient = RestClient.builder(
                new HttpHost("localhost", 9200)).build();
        // 创建 Elasticsearch 传输层，使用 Jackson 作为 JSON 映射器，用于处理数据的序列化和反序列化
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());
        // 创建 Elasticsearch 客户端，用于与 Elasticsearch 进行交互
        ElasticsearchClient client = new ElasticsearchClient(transport);

        // 连接 MySQL
        // 使用 JDBC 连接到本地 MySQL 数据库，需要提供数据库的 URL、用户名和密码
        Connection conn = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/your_database",
                "your_username",
                "your_password");
        // 创建一个 Statement 对象，用于执行 SQL 语句
        Statement stmt = conn.createStatement();
        // 执行 SQL 查询，从 user 表中获取所有数据
        ResultSet rs = stmt.executeQuery("SELECT * FROM user");

        // 遍历结果集，将每一行数据转换为 JSON 文档并导入到 Elasticsearch 中
        while (rs.next()) {
            // 创建一个 Map 对象，用于存储当前行的数据
            Map<String, Object> doc = new HashMap<>();
            // 获取结果集的元数据，包含列名、列类型等信息
            ResultSetMetaData metaData = rs.getMetaData();
            // 获取结果集中列的数量
            int columnCount = metaData.getColumnCount();
            // 遍历每一列，将列名和对应的值添加到 Map 中
            for (int i = 1; i <= columnCount; i++) {
                doc.put(metaData.getColumnName(i), rs.getObject(i));
            }

            // 创建一个索引请求，将 Map 转换为 JSON 文档并存储到 Elasticsearch 的指定索引中
            IndexRequest<Map<String, Object>> request = IndexRequest.of(i -> i
                    // 指定要存储文档的索引名称
                    .index("your_index")
                    // 设置要存储的文档内容
                    .document(doc)
                    // 刷新索引，确保文档立即可以被搜索到
                    .refresh(Refresh.True)
            );

            // 执行索引请求，将文档存储到 Elasticsearch 中
            IndexResponse response = client.index(request);
            // 打印存储文档的 ID，方便后续跟踪和调试
            System.out.println("Indexed document with ID: " + response.id());
        }

        // 关闭资源，释放数据库连接和 Elasticsearch 传输层
        rs.close();
        stmt.close();
        conn.close();
        transport.close();
    }

}
